package com.iwhalecloud.bss.kite.cucc.common.mq.producer;

import com.aliyun.openservices.ons.api.*;
import com.aliyun.openservices.ons.api.exception.ONSClientException;
import com.iwhalecloud.bss.kite.cucc.common.mq.serializer.FSTSerializer;
import com.iwhalecloud.bss.kite.cucc.common.mq.serializer.Serializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Properties;

/**
 * @author Linjj
 */
public class Producer {
    private static final Logger logger = LoggerFactory.getLogger(Producer.class);
    private static final Serializer FST_SERIALIZER = new FSTSerializer();

    public com.aliyun.openservices.ons.api.Producer producer;

    public static final String PROD_ENV = "prod";



    public Producer(String producerId, String accessKey, String secretKey, String sendMsgTimeoutMillis, String nameSrvAddr) {
        Properties properties = new Properties();
        properties.put(PropertyKeyConst.ProducerId, producerId);
        properties.put(PropertyKeyConst.AccessKey, accessKey);
        properties.put(PropertyKeyConst.SecretKey, secretKey);
        properties.setProperty(PropertyKeyConst.SendMsgTimeoutMillis, sendMsgTimeoutMillis);
        properties.put(PropertyKeyConst.NAMESRV_ADDR, nameSrvAddr);
        producer = ONSFactory.createProducer(properties);
        producer.start();
        logger.debug("MQ Producer Start!");
    }

    public Producer(String producerId, String accessKey, String secretKey, String sendMsgTimeoutMillis, String nameSrvAddr, String gid, String env) {
        Properties properties = new Properties();
        if (PROD_ENV.equals(env)) {
            properties.setProperty("GroupId", gid);
        }
        else {
            properties.setProperty("ProducerId", producerId);
        }
        properties.put(PropertyKeyConst.AccessKey, accessKey);
        properties.put(PropertyKeyConst.SecretKey, secretKey);
        properties.setProperty(PropertyKeyConst.SendMsgTimeoutMillis, sendMsgTimeoutMillis);
        properties.put(PropertyKeyConst.NAMESRV_ADDR, nameSrvAddr);
        producer = ONSFactory.createProducer(properties);
        producer.start();
        logger.debug("MQ Producer Start!");
    }

    public boolean send(String topic, String tag, Object sendObject, String key, boolean isOneway) {
        final Message msg = new Message();
        msg.setTopic(topic);
        msg.setTag(tag);
        msg.setKey(key);
        msg.setBody(FST_SERIALIZER.serialize(sendObject));
        /* 单向发送, 不关注是否成功*/
        if (isOneway) {
            producer.sendOneway(msg);
            logger.debug("MQ SendOneway - Topic:{} Tag:{}", msg.getTopic(), msg.getTag());
            return true;
        }

        /* 同步发送,关注结果 */
        try {
            SendResult sendResult = producer.send(msg);
            /* 同步发送消息，只要不抛异常就是成功 */
            if (sendResult != null) {
                logger.debug("MQ Send - Topic:{} Tag:{} MsgId:{}", msg.getTopic(), msg.getTag(), sendResult.getMessageId());
            }
            return true;
        }
        catch (Exception e) {
            /* 消息发送失败，需要进行重试处理，可重新发送这条消息或持久化这条数据进行补偿处理 */
            logger.error("MQ Send failed! Topic:{}", msg.getTopic());
//            e.printStackTrace();
        }
        return false;
    }

    public void start() {
        producer.start();
    }

    public boolean send(String topic, String tag, Object sendObject, String key) {
        return send(topic, tag, sendObject, key, false);
    }

    public boolean sendOneway(String topic, String tag, Object sendObject, String key) {
        return send(topic, tag, sendObject, key, true);
    }

    public boolean sendAsync(String topic, String tag, Object sendObject, String key) {
        final boolean[] resultFlag = {false};
        final ONSClientException[] onsClientException = new ONSClientException[1];
        final Message msg = new Message();
        msg.setTopic(topic);
        msg.setTag(tag);
        msg.setKey(key);
        msg.setBody(FST_SERIALIZER.serialize(sendObject));
        /* 异步发送消息*/
        producer.sendAsync(msg, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                if (sendResult != null) {
                    logger.debug("MQ Send - Topic:{} Tag:{} MsgId:{}", msg.getTopic(), msg.getTag(), sendResult.getMessageId());
                    resultFlag[0] = true;
                }
            }

            @Override
            public void onException(OnExceptionContext onExceptionContext) {
                onsClientException[0] = onExceptionContext.getException();
            }
        });
        if (!resultFlag[0]) {
            logger.error("MQ Send failed! Topic:{}", msg.getTopic());
        }
        return resultFlag[0];
    }

    public void shutdown() {
        producer.shutdown();
    }
}
